#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"

#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace bitmq
{

    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string &sip, int sport, AsyncWorker::ptr &worker) : _latch(1), _client(_worker->loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Client"),
                                                                                  _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
                                                                                  _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                                                                                  _worker(worker),
                                                                                  _channel_manager(std::make_shared<ChannelManager>())
        {
            // 注册业务请求处理函数
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse,
                                                                               this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse,
                                                                                this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));

            // 连接服务器--需要阻塞等待建立连接成功之后再返回

            _client.connect(); // 该函数会立刻返回，所以有可能没有成功建立连接成功
            _latch.wait();     // 阻塞等待，直到建立连接成功
        }

        Channel::ptr openChannel()
        {
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            // 向服务器发送创建信道的请求
            bool ret = channel->openChannel();
            if (ret == false)
            {
                DLOG("打开信道失败!")
                return Channel::ptr();
            }
            return channel;
        }
        void closeChannel(const Channel::ptr &channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }

    private:
        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp tt)
        {
            // 找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息！");
                return;
            }
            // 将得到的响应对象添加到信道的基础响应hashmap中
            // 一个连接有多个信道，一个信道可以有多个请求
            // 通过响应的信道ID,找到目标信道，再通过请求ID，将该应答存储到目标信道中
            channel->putBasicResponse(message);
        }
        void consumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            // 找到信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息！");
                return;
            }
            // 消息处理，封装成异步任务，交给线程池去处理
            _worker->pool.push([channel, message]()
                               { channel->consume(message); });
        }

        // 连接建立成功之后的回调函数，连接建立成功之后，唤醒上面的阻塞
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn = conn;
                _latch.countDown(); // 唤醒主线程中的阻塞
            }
            else
            {
                // 连接关闭时的操作
                _conn.reset();
            }
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                              const MessagePtr &message,
                              muduo::Timestamp tt)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

    private:
        muduo::net::TcpConnectionPtr _conn; // 客户端对应的连接
        AsyncWorker::ptr _worker;
        muduo::CountDownLatch _latch; // 实现同步的
        // loopthread一旦实例化就会创建线程直接启动事件监控

        muduo::net::TcpClient _client;  // 客户端
        ProtobufDispatcher _dispatcher; // 请求分发器
        ProtobufCodecPtr _codec;        // 协议处理器

        ChannelManager::ptr _channel_manager;
    };
}

#endif